package com.wing.config;

import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.shade.io.netty.util.concurrent.DefaultThreadFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static java.util.concurrent.Executors.newScheduledThreadPool;

@Component
public class PulsarClientInit {

    @Value("${pulsar.client.token}")
    private String token;
    @Value("${pulsar.client.serviceUrl}")
    private String serviceUrl;
    private  PulsarClient pulsars;

    private final ScheduledExecutorService executorService =
            newScheduledThreadPool(1, new DefaultThreadFactory("pulsar-cli-init"));

    public void init() {
        executorService.scheduleWithFixedDelay(this::initWithRetry, 0, 10, TimeUnit.SECONDS);
    }

    @PostConstruct
    private void initWithRetry() {
        try {
            PulsarClient pulsarClient = PulsarClient.builder()
                    .serviceUrl(serviceUrl)
                    .ioThreads(4)
                    .listenerThreads(10)
                    .operationTimeout(5, TimeUnit.SECONDS)
                    .connectionTimeout(15, TimeUnit.SECONDS)
                    .authentication(AuthenticationFactory.token(token))
                    .build();
            System.out.println("pulsar初始化client成功");
            this.setPulsars(pulsarClient);
            this.executorService.shutdown();
        } catch (Exception e) {
            System.out.println("pulsar初始化client异常, exception is "+e);
        }
    }

    public PulsarClient getPulsars() {
        return pulsars;
    }

    public void setPulsars(PulsarClient pulsars) {
        this.pulsars = pulsars;
    }

}
